package org.pupil.rpc.registry.registryImpl;

import com.alibaba.fastjson.JSON;
import org.pupil.rpc.common.RpcServiceNameBuilder;
import org.pupil.rpc.common.ServiceMeta;
import org.pupil.rpc.config.RpcProperties;
import org.pupil.rpc.registry.RegistryService;
import org.springframework.util.ObjectUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;


import java.io.IOException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * redis注册中心
 * <p>
 * 思路：
 * *      使用集合保存所有服务节点信息
 * * 服务启动：节点使用了redis作为注册中心后，将自身信息注册到redis当中(ttl：10秒)，并开启定时任务，ttl/2。
 * * 定时任务用于检测各个节点的信息，如果发现节点的时间 < 当前时间，则将节点踢出，如果没有发现，则续签自身节点
 * * 将节点踢出后，从服务注册表中找到对应key删除该节点的下的服务数据信息
 * *
 * * ttl :10秒
 * * 定时任务为ttl/2
 * *节点注册后启动心跳检测，检测服务注册的key集合，如果有服务到期，则删除,自身的服务则续签
 * * 服务注册后将服务注册到redis以及保存到自身的服务注册key集合，供心跳检测
 * *
 * * 如果有节点宕机，则其他服务会检测的，如果服务都宕机，则ttl会进行管理
 */
public class RedisRegistry implements RegistryService {

    // Jedis连接池，用于连接
    private JedisPool jedisPool;

    // 服务的UUID
    private String UUID;

    // 服务生存时间
    private static final int ttl = 10 * 1000; // 10s

    // 存储服务名
    private Set<String> serviceMap = new HashSet<>();

    // 创建一个单线程用于定时任务
    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();


    /**
     * 注册当前服务,将当前服务ip，端口，时间注册到redis当中，并且开启定时任务
     * 使用集合存储服务节点信息
     */
    public RedisRegistry() {
        // 获取配置信息
        RpcProperties properties = RpcProperties.getInstance();
        String[] split = properties.getRegisterAddr().split(":");
        // Jedis连接池配置类
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        // 设置连接池最大连接数
        poolConfig.setMaxTotal(10);
        // 设置连接池最大的空闲连接数
        poolConfig.setMaxIdle(5);
        // 创建Jedis连接池         连接池配置类，服务ip，服务端口
        jedisPool = new JedisPool(poolConfig, split[0], Integer.valueOf(split[1]));
        // 创建服务的uuid
        this.UUID = java.util.UUID.randomUUID().toString();
        // 健康监测
        heartbeat();
    }

    /**
     * 获取Jedis
     *
     * @return
     */
    private Jedis getJedis() {
        // 创建jedis
        Jedis jedis = jedisPool.getResource();
        RpcProperties properties = RpcProperties.getInstance();
        // 设置注册中心密码
        if (!ObjectUtils.isEmpty(properties.getRegisterPsw())) {
            jedis.auth(properties.getRegisterPsw());
        }
        return jedis;
    }

    /**
     * 健康监测
     * 每个服务节点可以进行自身续签
     * 同时有权利检测其他服务节点是否过期，过期则删除该服务
     */
    private void heartbeat() {
        int sch = 5;
        // 开启周期性定时任务
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            for (String key : serviceMap) {
                // 1.获取所有服务节点,查询服务节点的过期时间是否 < 当前时间。如果小于则有权将节点下的服务信息都删除
                // 保存的服务是同一份，获取多个是为了负载均衡
                List<ServiceMeta> serviceNodes = listServices(key);
                Iterator<ServiceMeta> iterator = serviceNodes.iterator();
                while (iterator.hasNext()) {
                    ServiceMeta node = iterator.next();
                    // 1.删除过期服务
                    if (node.getEndTime() < new Date().getTime()) {
                        iterator.remove();
                    }
                    // 2.自身续签
                    if (node.getUUID().equals(this.UUID)) {
                        node.setEndTime(node.getEndTime() + ttl / 2);
                    }
                }
                // 重新加载服务
                if (!ObjectUtils.isEmpty(serviceNodes)) {
                    loadService(key, serviceNodes);
                }
            }
            // 定时任务的执行时间 间隔为5秒
        }, sch, sch, TimeUnit.SECONDS);
    }

    /**
     * 加载服务
     *
     * @param key          服务信息
     * @param serviceMetas 所有的服务节点
     */
    private void loadService(String key, List<ServiceMeta> serviceMetas) {
        // 通过Lua脚本进行原子性操作
        String script = "redis.call('DEL',KEYS[1])\n" +
                "for i = 1, #ARGV do\n" +
                "   redis.call('RPUSH', KEYS[1], ARGV[i])\n" +
                "end \n" +
                "redis.call('EXPIRE', KEYS[1],KEYS[2])";
        List<String> keys = new ArrayList<>();
        keys.add(key); // 添加服务信息的键名
        keys.add(String.valueOf(10)); // 设置的过期时间
        List<String> values = serviceMetas.stream().map(o -> JSON.toJSONString(o)).collect(Collectors.toList());
        Jedis jedis = getJedis();
        jedis.eval(script, keys, values); // 执行脚本
        jedis.close();
    }

    /**
     * 获取所有服务节点
     *
     * @param key
     * @return
     */
    private List<ServiceMeta> listServices(String key) {
        Jedis jedis = getJedis();
        List<String> list = jedis.lrange(key, 0, -1);
        jedis.close();
        List<ServiceMeta> serviceMetas = list.stream().map(o -> JSON.parseObject(o, ServiceMeta.class)).collect(Collectors.toList());
        return serviceMetas;
    }


    /**
     * 服务注册
     *
     * @param serviceMeta
     * @throws Exception
     */
    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        // 创建服务名
        String key = RpcServiceNameBuilder.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion());
        // 不存在该服务名，添加该服务名
        if (!serviceMap.contains(key)) {
            serviceMap.add(key);
        }
        // 设置UUID
        serviceMeta.setUUID(this.UUID);
        // 设置超时时间
        serviceMeta.setEndTime(new Date().getTime() + ttl);
        // 获取jedis
        Jedis jedis = getJedis();
        String script = "redis.call('RPUSH', KEYS[1], ARGV[1])\n" +
                "redis.call('EXPIRE', KEYS[1], ARGV[2])";
        List<String> value = new ArrayList<>();
        value.add(JSON.toJSONString(serviceMeta));
        value.add(String.valueOf(10));
        jedis.eval(script, Collections.singletonList(key), value);
        jedis.close();
    }

    /**
     * 服务注销
     *
     * @param serviceMeta
     * @throws Exception
     */
    @Override
    public void unRegister(ServiceMeta serviceMeta) throws Exception {
        // 获取服务名
        String key = RpcServiceNameBuilder.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion());
        serviceMap.remove(key);
        // 获取jedis
        Jedis jedis = getJedis();
        String script = "redis.call('DEL',KEYS[1])";
        jedis.eval(script, 1, key);
        jedis.close();

    }

    /**
     * 获取 serviceName 下的所有服务
     *
     * @param serviceName
     * @return
     */
    @Override
    public List<ServiceMeta> discoveries(String serviceName) {
        return listServices(serviceName);
    }

    /**
     * 关闭
     *
     * @throws IOException
     */
    @Override
    public void destroy() throws IOException {
        jedisPool.close();
    }
}
